package com.zhuoyue;

import com.zhuoyue.utils.Log;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.Socket;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.function.Consumer;

/**
 * 管理socket
 *
 *  - 对socket进行封装,提供writer、read方法,尽量避免直接使用socket从而导致无法管理
 *
 *  - 如何判断socket是否断开？
 *      初步想的方式是通过判断 read方法返回-1或者writer方法出现异常，来判断socket是否断开。
 *      这样做确实可以判断出socket是否断开，但是存在线程安全问题。可能导致无限connec问题
 *      为了解决线程安全问题，我们考虑两种方式悲观锁和乐观锁：
 *      悲观锁的方式：
 *          1.会锁定writer线程(TimerThread)和read线程
 *      乐观锁：
 *          1.线程会做一部分无用操作。
 *
 *      综上：
 *          锁住线程，并不会影响用户线程，用户writer依然可以去插入队列。
 *          但是同样需要版本的方式，来控制，避免重复调用reconnect
 *
 *  - 重连
 *      - 要有一定策略，不能一直狂刷日志 (目前不做可配置，固定的策略)
 *
     - 提供连接状态 辅助writer时判断是否需要插入队列
 *      - connecting
 *      - connected
 *
 *   - 提供connection successful回调
 *
 *
 * ===============================================================================================
 *
 * @author Created by mr.liang
 */
public class ConnectionManagement {

    /** 配置管理 */
    private ClientConfig clientConfig;

    /** 每次成功都会调用*/
    private List<Consumer<Integer>> successCallable = new CopyOnWriteArrayList<>();

    /** 新的版本 */
    private volatile int newVersion;

    /** 旧的版本 ，当新旧版本不同时，表示在connecting*/
    private volatile int oldVersion;

    /** 嵌套字 */
    private volatile Socket socket;

    //输入流缓存
    private volatile InputStream ins;

    //输出流缓存
    private volatile OutputStream outs;

    /** 连接开始时间*/
    private long lastConnectTime ;


    public ConnectionManagement(ClientConfig clientConfig) {
        this.clientConfig = clientConfig;
        this.newVersion = 0;
        this.oldVersion = 0;
    }

    /**
     * 注册回调，每次succes成功的时候都会回调
     */
    public boolean registerCallback(Consumer<Integer> callable) {
        return successCallable.add(callable);
    }

    /**
     * 连接服务器
     *
     * 只能调用一次
     */
    public synchronized boolean connect() {
        if (oldVersion != 0) {
            return false;
        }
        reConnect(oldVersion);
        return true;
    }


    /**
     * 重连连接
     */
    private synchronized boolean reConnect(int version) {
        if (version != this.newVersion) {
            return true;
        }
        newVersion++;

        try {
            refresh();

            //回调列表
            for (Consumer<Integer> callable : successCallable){
                clientConfig.getThreadPoolExecutor().execute(()->{
                    try {
                        callable.accept(newVersion);
                    }catch (Exception e){
                        Log.error("connection回调异常: " +e.toString());
                        e.printStackTrace();
                    }
                });
            }
            return true;
        }catch (Exception e){
            Log.error("几乎不可能出现的异常: " +e.toString());
            e.printStackTrace();
            return false;
        } finally {
            oldVersion = newVersion;//version必须同步
        }
    }


    private void refresh() {
        //刷新socket

        if(socket!=null){
            try {
                ins.close();
            } catch (IOException e) { }
            try {
                outs.close();
            } catch (IOException e) { }
            try {
                socket.close();
            } catch (Exception e) { e.printStackTrace(); }
        }

        //防止连上就立刻断开的后台server,避免疯狂刷新日志
        long diff = System.currentTimeMillis() - lastConnectTime - clientConfig.getReconnectionMill();
        if(diff < 0){
            try {
                Thread.sleep(-diff);
            } catch (InterruptedException e) { }
        }
        lastConnectTime = System.currentTimeMillis();

        while (true){
            Log.info("init socket to access :  "+clientConfig.getServerIp() + ":"+clientConfig.getServerPort());
            try {
                socket = new Socket(clientConfig.getServerIp(),clientConfig.getServerPort());
                ins = socket.getInputStream();
                outs = socket.getOutputStream();
                Log.info( "successful connection, address : "+clientConfig.getServerIp() + ":"+clientConfig.getServerPort());
                return ;
            } catch (IOException e) {
                e.printStackTrace();
            }

            try {//打印顺序相关
                Thread.sleep(100);
            } catch (InterruptedException e) { }

            Log.info( " access error : " +clientConfig.getServerIp() + ":"+clientConfig.getServerPort()+ " , run again after waiting "
                    +clientConfig.getReconnectionMill()+" seconds");

            try {
                Thread.sleep(reConnectionPolicy(lastConnectTime));
            } catch (InterruptedException e) { }
        }
    }


    /**
     * 重连策略
     */
    protected int reConnectionPolicy(long callTime){
        long diff = System.currentTimeMillis() - callTime;
        //应该用二分法，但是考虑到代码的清晰就算了。
        if(diff < 3600_000 ){
            return clientConfig.getReconnectionMill();
        }else if(diff < 3600_000 * 8){
            return clientConfig.getReconnectionMill()<<1;
        }else if(diff < 3600_000 * 24){
            return clientConfig.getReconnectionMill()<<2;
        }else{
            return clientConfig.getReconnectionMill()<<3;
        }
    }

    public int read(byte[] b) throws IOException {
        int version = oldVersion;//这行的意思是，在这里version下的操作。

        int len = -1;
        while (true){
            try {
                len = ins.read(b);
            }catch (Exception e){
                e.printStackTrace();
            }
            if(len != -1){
                break;
            }
            reConnect(version);
            version = oldVersion;
        }

        return len;
    }

    public void writerAndFlush(byte[] bytes) throws IOException {
        int version = oldVersion;//这行的意思是，在这里version下的操作。

        while (true){
            try {
                outs.write(bytes);
                outs.flush();
                return ;
            }catch (Exception e){
                e.printStackTrace();
            }
            reConnect(version);
            version = oldVersion;
        }
    }

    public boolean isConnecting() {
        return newVersion != oldVersion;
    }


    public boolean isNotInitConnect() {
        return newVersion == 0;
    }
}
